package workerpool

import (
	"context"
	"fmt"
	"sync"
	"time"

	_err "github.com/romapres2010/goapp/pkg/common/error"
	_recover "github.com/romapres2010/goapp/pkg/common/recover"
)

// TaskState - статусы жизненного цикла task
type TaskState int

const (
	TASK_STATE_NEW                    TaskState = iota // task создан
	TASK_STATE_POOL_GET                                // task получен из pool
	TASK_STATE_POOL_PUT                                // task отправлен в pool
	TASK_STATE_READY                                   // task готов к обработкам
	TASK_STATE_IN_PROCESS                              // task выполняется
	TASK_STATE_DONE_SUCCESS                            // task завершился
	TASK_STATE_RECOVER_ERR                             // task остановился из-за паники
	TASK_STATE_TERMINATED_STOP_SIGNAL                  // task остановлен по причине получения сигнала об остановке
	TASK_STATE_TERMINATED_CTX_CLOSED                   // task остановлен по причине закрытия контекста
	TASK_STATE_TERMINATED_TIMEOUT                      // task остановлен по причине превышения timeout
)

// Task - содержимое задачи и результаты выполнения
type Task struct {
	parentCtx context.Context    // 父上下文，用于任务的创建
	ctx       context.Context    // 任务执行的上下文，可以用于取消任务
	cancel    context.CancelFunc // 用于取消 ctx 上下文的函数

	externalId  uint64             // 外部请求标识符，通常用于日志记录
	doneCh      chan<- interface{} // 用于通知外部世界任务完成的通道
	wg          *sync.WaitGroup    // sync.WaitGroup，用于等待任务完成。
	stopCh      chan interface{}   // 外部世界用来停止任务的通道
	localDoneCh chan interface{}   // 任务的本地完成信号通道，用于长时间运行的任务。

	id        uint64        // 任务编号，用于日志记录。
	state     TaskState     // 任务的当前状态。
	prevState TaskState     // 任务的前一个状态，用于状态转换的控制
	name      string        // 任务的名称，用于日志记录和监控。
	timeout   time.Duration // 任务的最大执行时间。
	timer     *time.Timer   // 超时计时器，用于超时停止任务。

	requests  []interface{} // 请求的输入数据，传递给处理函数。
	responses []interface{} // 处理请求的结果。
	err       error         // 处理请求时发生的错误。

	duration time.Duration // 任务实际执行的持续时间。

	f func(context.Context, context.Context, ...interface{}) (error, []interface{}) // 任务的处理函数，接受上下文和输入数据，并返回错误和结果数组

	mx sync.RWMutex // 读写锁，用于同步对任务状态的访问。
}

func NewTask(parentCtx context.Context, name string, doneCh chan<- interface{}, wg *sync.WaitGroup, id uint64, externalId uint64, timeout time.Duration, f func(context.Context, context.Context, ...interface{}) (error, []interface{}), requests ...interface{}) *Task {
	if f == nil || parentCtx == nil {
		return nil
	}

	// 从全局缓存中获取任务
	task := gTaskPool.getTask()

	// 不需要阻塞，从缓存中获取后，没有人再与之工作
	// 这可能意味着一旦任务从缓存中被取出，就不需要对它进行锁定或其他形式的同步控制，因为不会再有其他操作与之交互
	//task.mx.Lock()
	//defer task.mx.Unlock()
	// 不需要封锁，从缓存中获取后，没有人再使用它

	{ // 设定所有值到初始值，从缓存获取后
		task.parentCtx = parentCtx

		task.externalId = externalId
		task.doneCh = doneCh
		task.wg = wg

		task.id = id
		task.name = name
		task.timeout = timeout

		task.requests = requests
		task.responses = nil
		task.err = nil

		task.duration = 0

		task.f = f

		task.setStateUnsafe(TASK_STATE_READY) //准备处理
	} // 设定所有值到初始值，从缓存获取后

	return task
}

// DeleteUnsafe - 发送到缓存
func (ts *Task) DeleteUnsafe() {
	if ts != nil {
		gTaskPool.putTask(ts)
	}
}

// setState - 设置任务生命周期的状态
func (ts *Task) setState(state TaskState) {
	ts.mx.Lock()
	defer ts.mx.Unlock()
	ts.setStateUnsafe(state)
}

// setStateUnsafe - 设置任务生命周期的状态
func (ts *Task) setStateUnsafe(state TaskState) {
	ts.prevState = ts.state
	ts.state = state
}

// SetDoneChUnsafe - 外部世界可以设置一个通知完成的通道
func (ts *Task) SetDoneChUnsafe(doneCh chan<- interface{}) {
	ts.doneCh = doneCh
}

// SetWgUnsafe - 外部世界可以建立一个用于通知完成情况的渠道
func (ts *Task) SetWgUnsafe(wg *sync.WaitGroup) {
	ts.wg = wg
}

// GetExternalId - 获以Task的externalId
func (ts *Task) GetExternalId() uint64 {
	return ts.externalId
}

// GetError - 获取任务的错误信息
func (ts *Task) GetError() error {
	return ts.err
}

// SetError - 为任务设置错误信息
func (ts *Task) SetError(err error) {
	ts.err = err
}

// GetResponses - 获取Task的Requests
func (ts *Task) GetResponses() []interface{} {
	return ts.responses
}

// GetRequests - 获取Task的Requests
func (ts *Task) GetRequests() []interface{} {
	return ts.requests
}

// process - 从工作器(worker)启动任务(task)
func (ts *Task) process(workerID uint, workerTimeout time.Duration) {
	if ts == nil || ts.f == nil {
		return
	}

	// 在启动期间锁定任务，以排除同一指针的同时使用
	if ts.mx.TryLock() { // 不推荐使用 TryLock，但在这种情况下它非常方便
		defer ts.mx.Unlock()
	} else {
		//_log.Info("TASK - already locked for process: WorkerID, TaskId, TaskExternalId, TaskName", workerID, ts.id, ts.externalId, ts.name)
		ts.err = _err.NewTyped(_err.ERR_WORKER_POOL_TASK_ALREADY_LOCKED, ts.externalId, ts.name, ts.state, ts.prevState).PrintfError()
		return
	}

	// 检查只能启动准备好的任务
	if ts.state == TASK_STATE_READY {
		ts.setStateUnsafe(TASK_STATE_IN_PROCESS)
		//_log.Debug("START task: WorkerID, TaskId, TaskExternalId, TaskName", workerID, ts.id, ts.externalId, ts.name)
	} else {
		//_log.Info("TASK - has incorrect state to run: WorkerID, TaskId, TaskExternalId, TaskName", workerID, ts.id, ts.externalId, ts.name)
		ts.err = _err.NewTyped(_err.ERR_WORKER_POOL_TASK_INCORRECT_STATE, ts.externalId, ts.name, ts.state, ts.prevState, "READY").PrintfError()
		return
	}

	// 处理任务的panic情况
	defer func() {
		if r := recover(); r != nil {
			ts.err = _recover.GetRecoverError(r, ts.externalId, ts.name)
			ts.setStateUnsafe(TASK_STATE_RECOVER_ERR)
		}
	}()

	// 我们通知‘外部世界’任务完成，可以通过一个单独的通道或者使用 wg（WaitGroup）
	defer func() {
		// 可能会出现通道已关闭的情况，例如，如果‘外部世界’由于自身的超时原因没有等到我们，那么通道将已经被关闭了
		// 在编程中，特别是在涉及并发和通道（channel）的 Go 语言中，如果发送者尝试向一个已经关闭的通道发送数据，这将导致运行时恐慌（panic）。然而，如果接收者尝试从已关闭的通道接收数据，这是安全的，通道会正常返回其零值。了解这一点对于编写健壮的并发程序非常重要。
		if ts.doneCh != nil {
			ts.doneCh <- struct{}{}
		}

		// 如果在 WaitGroup 的范围内工作，那么将减少计数器
		if ts.wg != nil {
			ts.wg.Done()
		}
	}()

	if ts.timeout < 0 {
		// "Короткие" task (timeout < 0) не контролируем по timeout. Их нельзя прервать. Функция-обработчик запускается в goroutine worker
		ts.err, ts.responses = ts.f(ts.parentCtx, nil, ts.requests...)
		ts.setStateUnsafe(TASK_STATE_DONE_SUCCESS)
		return
	} else {
		// "Длинные" task запускаем в фоне и ожидаем завершения в отдельный локальный канал. Контролируем timeout

		var tic = time.Now() // временная метка начала обработки task

		go func() {
			defer func() {
				if r := recover(); r != nil {
					ts.err = _recover.GetRecoverError(r, ts.externalId, ts.name)
				}

				// Отправляем сигнал и закрываем канал, task не контролирует, успешно или нет завершился обработчик
				if ts.localDoneCh != nil {
					ts.localDoneCh <- struct{}{}
				}
			}()

			// Обработчик получает родительский контекст и локальный контекст task.
			// Локальный контекст task нужно контролировать в обработчике для определения необходимости остановки
			ts.err, ts.responses = ts.f(ts.parentCtx, ts.ctx, ts.requests...)
		}()

		// Определим, нужно ли контролировать timeout, ts.timeout имеет приоритет над workerTimeout
		var timeout time.Duration // предельное время работы task
		if ts.timeout > 0 {
			timeout = ts.timeout
		} else if workerTimeout > 0 {
			timeout = workerTimeout
		}

		// Если timeout == 0, то не контролировать timeout
		if timeout > 0 {
			// Task получает таймер всегда остановленным, сбрасывать канал таймера не требуется, так как он не сработал
			ts.timer.Reset(timeout) // Переставим таймер на новое значение
		}

		// Ожидаем завершения функции обработчика, наступления timeout или команды на закрытие task
		select {
		case <-ts.localDoneCh:
			if !ts.timer.Stop() { // остановим таймер
				<-ts.timer.C // Вероятность, что он сработал в промежутке между select из localDoneCh и выполнением ts.timer.Stop() крайне мала
			}

			ts.duration = time.Now().Sub(tic)
			ts.setStateUnsafe(TASK_STATE_DONE_SUCCESS)
			//_log.Debug("Task - DONE: WorkerID, TaskId, TaskExternalId, TaskName", workerID, ts.id, ts.externalId, ts.name)
			return
		case _, ok := <-ts.stopCh:
			if ok {
				// канал был открыт и получили команду на остановку
				//_log.Debug("Task - INTERRUPT - got stop signal: WorkerId, TaskExternalId, TaskName, WorkerTimeout", workerID, ts.externalId, ts.name, workerTimeout)
				ts.err = _err.NewTyped(_err.ERR_WORKER_POOL_STOP_SIGNAL, ts.externalId, fmt.Sprintf("[WorkerId='%v', TaskExternalId='%v', TaskName='%v', WorkerTimeout='%v']", workerID, ts.externalId, ts.name, workerTimeout))
				ts.setStateUnsafe(TASK_STATE_TERMINATED_STOP_SIGNAL)
			} else {
				// Не корректная ситуация с внутренней логикой - логируем для анализа
				//_log.Error("Task - INTERRUPT - stop chanel closed: WorkerId, TaskExternalId, TaskName, WorkerTimeout", workerID, ts.externalId, ts.name, workerTimeout)
			}
			// Закрываем локальный контекст task - функция обработчика должна корректно отработать это состояние и выполнить компенсационные воздействия
			if ts.cancel != nil {
				//_log.Debug("Task - close local context: WorkerID, TaskId, TaskExternalId, TaskName", workerID, ts.id, ts.externalId, ts.name)
				ts.cancel()
			}
			close(ts.localDoneCh)
			return
		case <-ts.timer.C:
			//_log.Info("Task - INTERRUPT - exceeded Timeout: WorkerId, TaskExternalId, TaskName, Timeout", workerID, ts.externalId, ts.name, timeout)
			ts.err = _err.NewTyped(_err.ERR_WORKER_POOL_TIMEOUT_ERROR, ts.externalId, ts.id, timeout).PrintfError()
			ts.setStateUnsafe(TASK_STATE_TERMINATED_TIMEOUT)
			// Закрываем локальный контекст task - функция обработчика должна корректно отработать это состояние и выполнить компенсационные воздействия
			if ts.cancel != nil {
				//_log.Debug("Task - close local context: WorkerID, TaskId, TaskExternalId, TaskName", workerID, ts.id, ts.externalId, ts.name)
				ts.cancel()
			}
			close(ts.localDoneCh)
			return

			// !!! вариант с time.After(ts.timeout) использовать нельзя, так как будут оставаться "повешенными" таймеры
			//case <-time.After(ts.timeout):
			//	//_log.Info("Task - INTERRUPT - exceeded TaskTimeout: WorkerId, TaskExternalId, TaskName, TaskTimeout", workerID, ts.externalId, ts.name, ts.timeout)
			//	ts.err = _err.NewTyped(_err.ERR_WORKER_POOL_TIMEOUT_ERROR, ts.externalId, ts.timeout).PrintfError()
			//	ts.setState(TASK_STATE_TERMINATED_TIMEOUT)
			//	return
			// !!! вариант с time.After(ts.timeout) использовать нельзя, так как будут оставаться "повешенными" таймеры

			// !!! Контроль за контекстом - дорогая операция
			//case <-ts.ctx.Done():
			//	//_log.Info("Task - STOP - got context close: WorkerId, TaskExternalId", workerID, ts.externalId)
			//	ts.err = _err.NewTyped(_err.ERR_WORKER_POOL_CONTEXT_CLOSED, ts.externalId, "Worker pool - Task - got context close")
			//	ts.setStateUnsafe(TASK_STATE_TERMINATED_CTX_CLOSED)
			//	return
			// !!! Контроль за контекстом - дорогая операция
		}
	}
}

// Stop - принудительная остановка task
func (ts *Task) Stop() {

	// Останавливать можно только в определенных статусах
	if ts.state == TASK_STATE_NEW || ts.state == TASK_STATE_IN_PROCESS || ts.state == TASK_STATE_DONE_SUCCESS {
		//_log.Debug("TASK  - send STOP signal: TaskExternalId, TaskName", ts.externalId, ts.name)
		// Отправляем сигнал и закрываем канал
		if ts.stopCh != nil {
			ts.stopCh <- true
			close(ts.stopCh)
		}
	}
}
